package edu.zju.udms.dnavigator.entities.kettle;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.log4j.Logger;

import edu.zju.udms.dnavigator.core.entity.Entity;
import edu.zju.udms.dnavigator.core.entity.ExtractorField;
import edu.zju.udms.dnavigator.core.entity.FieldTypeEnum;
import edu.zju.udms.dnavigator.core.entity.Record;
import edu.zju.udms.dnavigator.core.entity.Relation;
import edu.zju.udms.dnavigator.core.extract.Extractor;
import edu.zju.udms.dnavigator.core.utils.Tuple3;
import edu.zju.udms.dnavigator.entities.mongodb.MongoDBExtractor;
import edu.zju.udms.dnavigator.entities.mysql.MySQLExtractor;

public class KettleDBRepositoryExtractor extends Extractor{
	private static final Logger logger = Logger.getLogger(KettleDBRepositoryExtractor.class);
	@ExtractorField(FieldTypeEnum.WEB_USE)
	public static final String SOURCE="Kettle";

	@ExtractorField(FieldTypeEnum.SAVE)
	@ExtractorField(FieldTypeEnum.WEB_USE)
	String url;

	@ExtractorField(FieldTypeEnum.SAVE)
	String driverName;

	@ExtractorField(FieldTypeEnum.SAVE)
	@ExtractorField(FieldTypeEnum.WEB_USE)
	String logUrl;
	
	@ExtractorField(FieldTypeEnum.SAVE)
	@ExtractorField(FieldTypeEnum.SHOW_IN_PANEL)
	@ExtractorField(FieldTypeEnum.WEB_USE)
	String stepTable;
	
	@ExtractorField(FieldTypeEnum.SAVE)
	@ExtractorField(FieldTypeEnum.SHOW_IN_PANEL)
	@ExtractorField(FieldTypeEnum.WEB_USE)
	String transformTable;
	
	@ExtractorField(FieldTypeEnum.SAVE)
	@ExtractorField(FieldTypeEnum.SHOW_IN_PANEL)
	@ExtractorField(FieldTypeEnum.WEB_USE)
	String sourceKey;
	

	int logSqlPosition = 0;
	
	@Override
	public Tuple3<List<Entity>, List<Relation>, List<Record>> extract() throws Exception {
		logger.info("extract time: " + logSqlPosition);

		List<Entity> entities = new ArrayList<>();
		List<Relation> relations = new ArrayList<>();
		int tableInputTypeId = -1;
		int tableOutputTypeId = -1;
		int mongoDBInputTypeId = -1;
		int mongoDBOutputTypeId = -1;
		Class.forName(driverName);
		logger.info("driver name: "+driverName+"; jdbc url : "+url);
		
		Connection conn = DriverManager.getConnection(url);
		Statement stmt = conn.createStatement();
		Statement tempStmt = conn.createStatement();
		
		Connection logConn = DriverManager.getConnection(logUrl);
		Statement logStmt = logConn.createStatement();
		
		//transformation
		Map<Integer,KettleTransform> trans = new HashMap<>();
		Map<Integer,KettleStep> steps = new HashMap<>();
		
		Map<String, Integer> transNameId = new HashMap<>();
		
		// stepNameId 中Key为transform的Id，结果为该transform的step的名字和id构成的HashMap
		Map<Integer, HashMap<String, Integer>> stepNameId = new HashMap<>();
		
		ResultSet res = stmt.executeQuery("select * from R_TRANSFORMATION");
		
		while(res.next()) {
			int id = res.getInt("ID_TRANSFORMATION");
			String name = res.getString("NAME");
			String desc = res.getString("DESCRIPTION");
			String owner = res.getString("CREATED_USER");
			Timestamp createTime = res.getTimestamp("CREATED_DATE");
			Timestamp updateTime = res.getTimestamp("MODIFIED_DATE");
			KettleTransform tran = new KettleTransform(
					Integer.toString(id), name, sourceKey,
					createTime!=null?createTime.getTime():0,
					updateTime!=null?updateTime.getTime():0,
					owner, desc);
			String query = "select count(*) from  "+transformTable+" where TRANSNAME=\""+name+"\"";
			ResultSet logRes = logStmt.executeQuery(query);
			logRes.next();
			tran.runTime = logRes.getInt(1);
			
			entities.add(tran);
			trans.put(id, tran);
			transNameId.put(name, id);
			stepNameId.put(id, new HashMap<>());
		}
		
		// Find the step type id from database
		res = stmt.executeQuery("select * from R_STEP_TYPE");
		while(res.next()) {
			int id = res.getInt("ID_STEP_TYPE");
			String code = res.getString("CODE");
			switch(code){
			case "TableInput":
				tableInputTypeId=id;
				break;
			case "TableOutput":
				tableOutputTypeId=id;
				break;
			case "MongoDbInput":
				mongoDBInputTypeId=id;
				break;
			case "MongoDbOutput":
				mongoDBOutputTypeId=id;
				break;
			}
		}
		
		//step
		res = stmt.executeQuery("select * from R_STEP");
		while(res.next()) {
			int id = res.getInt("ID_STEP");
			int tid = res.getInt("ID_TRANSFORMATION");
			int typeId = res.getInt("ID_STEP_TYPE");
			String name = res.getString("NAME");
			String desc = res.getString("DESCRIPTION");
			KettleStep step = new KettleStep(Integer.toString(id), name, sourceKey, 
					desc, trans.get(tid).getId().toString());
			if(typeId==tableInputTypeId || typeId==tableOutputTypeId) {
				step.input = (typeId == tableInputTypeId);
				step.dbSource = MySQLExtractor.SOURCE;
				ResultSet tempRes = tempStmt.executeQuery("SELECT * FROM R_STEP_ATTRIBUTE where ID_STEP="+id+" and code='table'");
				if(tempRes.next()) {
					step.dbTable = tempRes.getString("VALUE_STR");
				} else {
					// 通过解析sql语句查找该步骤操作的数据表
					// TODO 换成正则表达式
					tempRes = tempStmt.executeQuery("SELECT * FROM R_STEP_ATTRIBUTE where ID_STEP="+id+" and code='sql'");
					if (tempRes.next()) {
						String sql = tempRes.getString("VALUE_STR");
						int fromIndex = sql.indexOf("FROM");
						String strToGetTable = sql.substring(fromIndex+4).trim().replace('\r', ' ').replace('\n', ' ');
						int endIndex = strToGetTable.indexOf(' ');
						step.dbTable = endIndex < 0? strToGetTable : strToGetTable.substring(0, endIndex);
					}
				}
				tempRes = tempStmt.executeQuery("SELECT * FROM R_STEP_DATABASE where ID_STEP="+id);
				int db_id = 0;
				if(tempRes.next()) {
					db_id = tempRes.getInt("ID_DATABASE");
				}
				tempRes = tempStmt.executeQuery("SELECT * FROM R_DATABASE where ID_DATABASE="+db_id);
				if(tempRes.next()) {
					String hostName = tempRes.getString("HOST_NAME");
					step.dbHost = hostName.equals("localhost") ? "127.0.0.1" : hostName;
					step.dbPort = Integer.toString(tempRes.getInt("PORT"));
					step.dbDatabase = tempRes.getString("DATABASE_NAME");
				}
			} else if(typeId == mongoDBInputTypeId || typeId == mongoDBOutputTypeId) {
				ResultSet tempRes = tempStmt.executeQuery("SELECT * FROM R_STEP_ATTRIBUTE where ID_STEP="+id);
				step.input = (typeId == mongoDBInputTypeId);
				step.dbSource = MongoDBExtractor.SOURCE;
				while(tempRes.next()) {
					String tcode = tempRes.getString("CODE");
					String tvalue = tempRes.getString("VALUE_STR");
					if(step.input) {
						switch (tcode) {
						case "hostname":
							step.dbHost = tvalue.equals("localhost") ? "127.0.0.1" : tvalue;
							break;
						case "port":
							step.dbPort = tvalue;
							break;
						case "db_name":
							step.dbDatabase = tvalue;
							break;
						case "collection":
							step.dbTable = tvalue;
							break;
						}
					} else {
						switch (tcode) {
						case "mongo_host":
							step.dbHost = tvalue.equals("localhost") ? "127.0.0.1" : tvalue;
							break;
						case "mongo_port":
							step.dbPort = tvalue;
							break;
						case "mongo_db":
							step.dbDatabase = tvalue;
							break;
						case "mongo_collection":
							step.dbTable = tvalue;
							break;
						}
					}
				}
			}
			entities.add(step);
			steps.put(id, step);
			stepNameId.get(tid).put(name, id);
		}
		
		// hop
		res = stmt.executeQuery("select * from R_TRANS_HOP");
		while(res.next()) {
			int stepFrom = res.getInt("ID_STEP_FROM");
			int stepTo = res.getInt("ID_STEP_TO");
			relations.add(new Relation(steps.get(stepFrom).getId().toString(),steps.get(stepTo).getId().toString(),"next"));
		}
		
		// kettle log -> records
		List<Record> records = new ArrayList<Record>();

		// transform log -> records
		res = logStmt.executeQuery("select * from " + transformTable);
		while (res.next()) {
			int transId = transNameId.get(res.getString("TRANSNAME"));
			records.add(new KettleTransformRecord(sourceKey, transId,
					DateFormatUtils.format(res.getTimestamp("LOGDATE"), "yyyy-MM-dd hh:mm:ss"),	
					res.getInt("LINES_READ"), res.getInt("LINES_WRITTEN"), res.getInt("LINES_UPDATED"),
					res.getInt("LINES_INPUT"), res.getInt("LINES_OUTPUT"), res.getInt("LINES_REJECTED"), res.getInt("ERRORS")));		
		}

		// step log -> records		
		res = logStmt.executeQuery("select * from " + stepTable);
		while (res.next()) {
			int transId = transNameId.get(res.getString("TRANSNAME"));
			String stepName = res.getString("STEPNAME");
			records.add(new KettleStepRecord(sourceKey, transId,
					stepNameId.get(transId).get(stepName),
					DateFormatUtils.format(res.getTimestamp("LOG_DATE"), "yyyy-MM-dd hh:mm:ss"),	
					res.getInt("LINES_READ"), res.getInt("LINES_WRITTEN"), res.getInt("LINES_UPDATED"),
					res.getInt("LINES_INPUT"), res.getInt("LINES_OUTPUT"), res.getInt("LINES_REJECTED"), res.getInt("ERRORS")));		
		}
		
		conn.close();
		logConn.close();

		return new Tuple3<List<Entity>, List<Relation>, List<Record>>(entities, relations, records);
	}

	@Override
	public String getSource() {
		return SOURCE;
	}

}
